<?php
/**
 *
 * 帖子更新
 *
 * TopicUpdate.class.php
 * Created by IntelliJ IDEA.
 *
 * @author: hjh <hjh@jf.com>
 *
 * Date: 2017/10/30
 * Time: 19:46
 *
 */

require_once __DIR__ . '/Event.class.php';
require QUEUE_ROOT . 'lib/ElasticSearch.class.php';

class MeiteTopicUpdate extends Queue
{
    const TRIGGER_COUNT = 10;
    const SAVE_INTERVAL = 20; //秒

    const MYSQL_DB = 'meite_circle';
    const TABLE_NAME = 'Topic';
    const REDIS_GROUP = 'other';
    const REDIS_DB = 0;
    const REDIS_ARTICLE_KEY = 'MEITE_TOPIC_ID:%d'; // topicId

    //事件队列 $topicId=.
    private $eventQueue;

    public function __construct()
    {
        $this->eventQueue = array();
    }

    public function run($data)
    {
        try {
            $data = unserialize($data);
            is_array($data) || $data = [];

            if (empty($data['TopicId'])) {
                return;
            }

            $topicId = $data['TopicId'];
            if (isset($this->eventQueue[$topicId])) {
                $event = $this->eventQueue[$topicId];
            } else {
                $event = new Event($topicId);
            }

            $event->triggerCount++;

            $now = time();
            if ($this->needUpdate($event, $now)) {

                // 参考Constants RedisDBConstant
                $redis = $this->getRedis(self::REDIS_GROUP, self::REDIS_DB);
                $pdo = $this->getPDO(self::MYSQL_DB);

                try {
                    $this->update($redis, $pdo, $event);
                    unset($this->eventQueue[$topicId]);
                } catch (Exception $e) {
                    $this->log(__CLASS__, array('error:' . $e->getMessage(), 'time: ' . date("Y-m-d H:i:s")));

                } finally {
                    $pdo = null;
                    if ($redis) {
                        $redis->close();
                        $redis = null;
                    }
                }
            } else {
                $this->eventQueue[$topicId] = $event;
            }
        } catch (Exception $e) {
            $this->log(__CLASS__, array('error:' . $e->getMessage(), "time:" . date("Y-m-d H:i:s")));
        }
        gc_collect_cycles();
    }

    /**
     * @param \Redis $redis
     * @param \PDO   $pdo
     * @param Event  $event
     * @return bool
     */
    private function update($redis, $pdo, $event)
    {

        //根据文章ID 从redis中获取对应的 数据
        $redisKey = sprintf(self::REDIS_ARTICLE_KEY, $event->id);
        $data = $redis->hGetAll($redisKey);
        if (empty($data)) {
            $this->log(__CLASS__, array('error: ' . " key[{$redisKey}] not found in db, time:" . date('Y-m-d H:i:s')));
            return false;
        }
        $columns = ['ViewCount', 'CommentCount', 'LikeCount', 'DislikeCount', 'ShareCount'];
        $selectFields = implode(', ', $columns);
        $sql = sprintf("select %s, `UpdateTime` from `%s` WHERE `TopicId` = '%d'", $selectFields, self::TABLE_NAME, $event->id);
        $query = $pdo->prepare($sql);
        $r = $query->execute();
        if (empty($r)) {
            $this->log(__CLASS__, array('select UpdateTime error, sql: ' . $sql . ' time:' . date('Y-m-d H:i:s')));
            return false;
        }
        $res = $query->fetch(PDO::FETCH_ASSOC);
        $updateTime = isset($res['UpdateTime']) ? $res['UpdateTime'] : date('Y-m-d H:i:s');

        if ($res) {
            //如果缓存中没有，那么就从数据库读取
            foreach ($res as $col => $val) {
                if (isset($data[$col])) {
                    continue;
                }
                $data[$col] = $val;
            }
        }

        $res = null;
        $query = null;
        $r = null;

        //更新mysql
        $fields = [];
        unset($data['UpdateTime']);
        unset($data['Hot']);

        $keys = array_keys($data);
        foreach ($keys as $key) {
            if (empty($data[$key])) {
                continue;
            }
            $fields[] = "`$key`='{$data[$key]}'";
        }

        //update Topic set ViewCount=xx, CommentCount=xx, LikeCount=xx, DisLikeCount=xx, HotScore=xx where TopicId = 1
        $sql = sprintf("UPDATE `%s` SET %s WHERE `TopicId` = '%d'", self::TABLE_NAME, implode(', ', $fields), $event->id);
        $query = $pdo->prepare($sql);
        $r = $query->execute();
        if ($r === false) {
            $this->log(__CLASS__, array('sql:' . $sql, " time:" . date("Y-m-d H:i:s")));
        }

        $r = null;
        $query = null;

        $data['HotScore'] = $this->score($data, $updateTime);
        $this->updateEs($event->id, $data);
        return $r;
    }

    /**
     * 计算评分。 同Recommend模块中的 HotRank 计算方法
     *
     * @param $info
     * @param updateTime
     *
     * @return double
     */
    private function score($info, $updateTime)
    {
        $viewCount = isset($info['ViewCount']) ? $info['ViewCount'] : 0;
        $likeCount = isset($info['LikeCount']) ? $info['LikeCount'] : 0;
        $disLikeCount = isset($info['DisLikeCount']) ? $info['DisLikeCount'] : 0;
        $shareCount = isset($info['ShareCount']) ? $info['ShareCount'] : 0;
        $commentCount = isset($info['CommentCount']) ? $info['CommentCount'] : 0;

        $T = (time() - strtotime($updateTime)) / 7200;
        //$T = max($T, 48);
        $z = 0.04;
        $s = 1 * $viewCount + 5 * ($likeCount - $disLikeCount) + 5 * $shareCount + 2 * $commentCount;
        $score = doubleval(log(max($s, 1))) * doubleval(exp(-1 * $z * max($T, 0)));

        $this->log(__CLASS__, 'score: ' . json_encode($info) . ', s:' . $s . ', HotScore:' . $score . ', UpdateTime:' . $updateTime);
        return $score;
    }

    private function needUpdate($event, $now)
    {
        return $event->triggerCount >= self::TRIGGER_COUNT || $now - $event->lastUpdateTime >= self::SAVE_INTERVAL;
    }

    private function updateAll($force = false)
    {
        $pdo = null;
        $redis = null;

        try {
            $unset_ids = [];
            $now = time();
            $pdo = $this->getPDO(self::MYSQL_DB);
            $redis = $this->getRedis(self::REDIS_GROUP, self::REDIS_DB);

            foreach ($this->eventQueue as $id => $event) {
                if ($force || $this->needUpdate($event, $now)) {
                    $this->update($redis, $pdo, $event);
                    $unset_ids[] = $id;
                }
            }

            if (count($unset_ids) > 0) {
                foreach ($unset_ids as $id) {
                    unset($this->eventQueue[$id]);
                }
            }
            unset($unset_ids);
        } catch (Exception $e) {
            $this->log(__CLASS__, array('error:' . $e->getMessage(), "time:" . date("Y-m-d H:i:s")));

        } finally {
            $pdo = null;

            if (isset($redis)) {
                $redis->close();
                $redis = null;
            }
            gc_collect_cycles();
        }
    }

    /**
     * 更新ES 评分
     *
     * @param $topicId
     * @param $data
     */
    private function updateEs($topicId, $data)
    {
        $query = "meite_topic/list/{$topicId}/_update";

        $filedMap = [
            'CommentCount' => ['int', 'commentcount'],
            'LikeCount' => ['int', 'likecount'],
            'DisLikeCount' => ['int', 'dislikecount'],
            'HotScore' => ['double', 'hotscore'],
            'ViewCount' => ['int', 'viewcount'],
            'ShareCount' => ['int', 'sharecount'],
        ];
        $doc = [];
        foreach ($data as $key => $value) {
            if (!isset($filedMap[$key])) {
                continue;
            }
            $info = $filedMap[$key];
            switch ($info[0]) {
                case 'int':
                    $value = intval($value);
                    break;
                case 'double':
                    $value = (double)$value;
                    break;
            }
            $doc[$info[1]] = $value;
        }

        $this->log(__CLASS__, "updateTopicEs: {$topicId}, " . json_encode($doc, JSON_UNESCAPED_UNICODE));
        $es = $this->getES();

        $ret = $es->query($query, 'POST', [
            'doc' => $doc
        ]);
        if ($ret === false) {
            $this->log(__CLASS__, array('error: topicId:' . $topicId, "time:" . date("Y-m-d H:i:s") . ' error:' . json_encode($ret, JSON_UNESCAPED_UNICODE)));
        }
    }

    /**
     * 队列未取到数据时的活动信号传递
     *
     * @see Queue::activeSignal()
     */
    public function activeSignal()
    {
        $this->updateAll();
    }

    /**
     * (non-PHPdoc)
     *
     * @see Queue::shutDownSignal()
     */
    public function shutDownSignal()
    {
        $this->updateAll(true);
        $this->log('QueueKillInfo', [
            'Killed',
            __CLASS__,
            date("Y-m-d H:i:s"),
            $this->__lastdata
        ]);
        exit;
    }
}
